package com.uw.uwrpc.core.registry.impl;

import com.alibaba.fastjson.JSON;
import com.uw.uwrpc.core.api.RegistryCenter;
import com.uw.uwrpc.core.config.UWRpcProperty;
import com.uw.uwrpc.core.constant.UWConstant;
import com.uw.uwrpc.core.meta.InstanceMeta;
import com.uw.uwrpc.core.meta.ServiceMeta;
import com.uw.uwrpc.core.registry.ChangedListener;
import com.uw.uwrpc.core.registry.RegistryCenterChangedEvent;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.CollectionUtils;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
public class ZookeeperRegistryCenter implements RegistryCenter, ApplicationContextAware {

    private CuratorFramework client = null;
    private ApplicationContext applicationContext = null;


    @Override
    public void start() {
        UWRpcProperty uwRpcProperty = applicationContext.getBean(UWRpcProperty.class);

        client = CuratorFrameworkFactory.builder()
                .connectString(uwRpcProperty.getZk().getServer())
                .namespace(uwRpcProperty.getZk().getRoot())
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .sessionTimeoutMs(5000)
                .build();

        client.start();
        log.debug(" ===> zk client started.");
    }

    @Override
    public void stop() {
        client.close();
        log.debug(" ===> zk client stopped.");
    }

    @Override
    @SneakyThrows
    public void register(ServiceMeta serviceMeta, InstanceMeta instanceMeta) {
        String servicePath = "/" + serviceMeta.serviceSign();
        // 创建服务持久化节点
        if (client.checkExists().forPath(servicePath) == null) {
            client.create().withMode(CreateMode.PERSISTENT).forPath(servicePath, serviceMeta.toMetas().getBytes(StandardCharsets.UTF_8));
        }
        // 创建实例临时节点
        String instancePath = servicePath + "/" + instanceMeta.instanceSign();
        client.create().withMode(CreateMode.EPHEMERAL).forPath(instancePath, instanceMeta.toMetas().getBytes(StandardCharsets.UTF_8));
        log.debug("register service: {}, instance:{}", serviceMeta, instanceMeta);
    }

    @Override
    @SneakyThrows
    public void unregister(ServiceMeta serviceMeta, InstanceMeta instanceMeta) {
        String servicePath = "/" + serviceMeta.serviceSign();
        // 不存在服务，直接返回
        if (client.checkExists().forPath(servicePath) == null) {
            log.debug("servicePath not exist with service: {}, instance:{}", serviceMeta, instanceMeta);
            return;
        }
        // 删除实例临时节点
        String providerPath = servicePath + "/" + instanceMeta.instanceSign();
        client.delete().forPath(providerPath);
        log.debug("unregister service: {}, instance:{}", serviceMeta, instanceMeta);
    }

    @Override
    @SneakyThrows
    public List<InstanceMeta> fetchAll(ServiceMeta serviceMeta) {
        String servicePath = "/" + serviceMeta.serviceSign();
        // 不存在服务，直接返回
        if (client.checkExists().forPath(servicePath) == null) {
            return new ArrayList<>();
        }

        // 返回服务下所有实例节点
        List<String> nodeList = client.getChildren().forPath(servicePath);
        log.debug(" ===> fetchAll nodeList: " + nodeList);
        return nodeList.stream().map(e -> getInstanceMeta(e, servicePath)).collect(Collectors.toList());
    }

    @NotNull
    private InstanceMeta getInstanceMeta(String instanceSign, String servicePath) {
        String host = instanceSign.split(UWConstant.SEPARATOR)[0];
        Integer port = Integer.valueOf(instanceSign.split(UWConstant.SEPARATOR)[1]);
        InstanceMeta instanceMeta = InstanceMeta.http(host, port);

        String instancePath = servicePath + "/" + instanceMeta.instanceSign();
        try {
            byte[] bytes = client.getData().forPath(instancePath);
            Map<String, Object> parameters = JSON.parseObject(new String(bytes));
            if (!CollectionUtils.isEmpty(parameters)) {
                for (Map.Entry<String, Object> entry : parameters.entrySet()) {
                    String key = entry.getKey();
                    String value = entry.getValue() == null ? "" : String.valueOf(entry.getValue());
                    log.debug("instance:{}, parameter key={}, value={}", instanceMeta.instanceSign(), key, value);
                    instanceMeta.getParameters().put(key, value);
                }
            }
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }

        return instanceMeta;
    }


    @Override
    public void subscribe(ServiceMeta serviceMeta, ChangedListener listener) {
        String servicePath = "/" + serviceMeta.serviceSign();

        CuratorCache curatorCache = CuratorCache.builder(client, servicePath).build();
        CuratorCacheListener cacheListener = CuratorCacheListener.builder().forCreatesAndChanges((ChildData oldNode, ChildData node) -> {
            log.debug(" ==> forCreatesAndChanges");
            List<InstanceMeta> instanceMetaList = fetchAll(serviceMeta);
            listener.onChange(new RegistryCenterChangedEvent(instanceMetaList));
        }).forDeletes((childData) -> {
            log.debug(" ==> forDeletes");
            List<InstanceMeta> instanceMetaList = fetchAll(serviceMeta);
            listener.onChange(new RegistryCenterChangedEvent(instanceMetaList));
        }).build();
        curatorCache.listenable().addListener(cacheListener);
        curatorCache.start();
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
